package com.intct.flink.project.sql;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-10-07 07:40
 * 测试数据：
 * INSERT INTO `m1`.`test01` (`id`, `name`, `update_time`, `price`) VALUES (19, '李四', 1111111, 1000.00000);
 *
 * update `m1`.`test01` set price = 500 where id = 18
 *
 * INSERT INTO `m1`.`test01` (`id`, `name`, `update_time`, `price`) VALUES (20, '王五', 22222, 1000.00000);
 *
 * delete from  `m1`.`test01` where id = 19;
 *
 * update `m1`.`test01` set price = 100 where id = 18
 *
 * INSERT INTO `m1`.`test01` (`id`, `name`, `update_time`, `price`) VALUES (21, '赵六', 33333333, 1000.00000);
 */
public class OldTumbleWindowTestApp {
    public static void main(String[] args) {
//        Configuration config = new Configuration();
//        config.set(CheckpointingOptions.CHECKPOINT_STORAGE, "filesystem");
//        config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, "file://d:/test/chk/");

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/test/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        TableConfig config = tenv.getConfig();
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled",true);
        config.getConfiguration().setString("table.exec.emit.early-fire.delay","60000 ms");

        tenv.executeSql(
                "CREATE TABLE order_mysql (    " +
                        " id            INT ,             " +
                        " name          STRING ,          " +
                        " price         DECIMAL(10, 5),   " +
                        " create_time   timestamp(3)  ,      " +
                        " update_time   bigint  ,      " +
                        " rt AS create_time           ,   "+
                        " proc_time AS proctime(),        " +
//                        " rt AS to_timestamp_ltz(update_time,3),   " +
                        " watermark for rt as rt - interval '0' second ,   " +
//                        " watermark for create_time as create_time - interval '0' second ,   " +
                        "     PRIMARY KEY (id) NOT ENFORCED         " +
                        "     ) WITH (                              " +
                        "     'connector' = 'mysql-cdc',            " +
                        "     'hostname' = 'cdh-node'   ,            " +
                        "     'port' = '13306'          ,            " +
                        "     'username' = 'root'      ,            " +
                        "     'password' = 'Test_090110'      ,            " +
                        "     'scan.startup.mode' = 'latest-offset', " +
                        "     'database-name' = 'm1' ,            " +
                        "     'table-name' = 'test01'            " +
                        ")"
        );

//        tenv.executeSql("select * from order_mysql").print();

        tenv.executeSql(
                "select                               " +
//                        " TUMBLE_START(rt, INTERVAL '10' SECOND) AS window_start" +
//                        " , TUMBLE_END(rt, INTERVAL '10' SECOND) AS window_end" +
                        " TUMBLE_START(rt, INTERVAL '24' HOUR) AS window_start" +
                        " , TUMBLE_END(rt, INTERVAL '24' HOUR) AS window_end" +
                        " , PROCTIME() AS  update_time " +
                        " , sum(price)  price            " +
                   " from order_mysql " +
                   " GROUP BY" +
//                        " TUMBLE(rt, INTERVAL '10' SECOND)")
                        " TUMBLE(rt, INTERVAL '24' HOUR)")
                .print();
//        tenv.executeSql("select unix_timestamp(cast(create_time as string), 'yyyy-MM-dd hh:mm:ss') from order_mysql").print();
    }
}